#include "TcpConnection.h"
#include "Logger.h"
#include "Socket.h"
#include "Channel.h"
#include "EventLoop.h"


#include <cerrno>
#include <sys/types.h>
#include <sys/socket.h>
#include <string.h>
#include <strings.h>
#include <netinet/tcp.h>
#include <sys/socket.h>


static EventLoop* CheckLoopNotNull(EventLoop* loop)
{
    if (loop == nullptr) {
        LOG_FATAL("TcpConnection Loop is null...... \n");
    }

    return loop;
}

TcpConnection::TcpConnection(EventLoop *loop
        , const std::string &nameArg
        , int sockfd
        , const InetAddress &localAddr
        , const InetAddress &perrAddr)
    : m_loop(CheckLoopNotNull(loop))        
    , m_name(nameArg)
    , m_state(kConnecting)
    , m_reading(true)
    , m_socket(new Socket(sockfd))
    , m_channel(new Channel(loop, sockfd))
    , m_localAddr(localAddr)
    , m_peerAddr(perrAddr)
    , m_highWaterMark(64 * 1024 * 1024) // g4M
{
    // 给channel设置相应的回调函数，poller会给channel通知它所感兴趣的事件是否发生了，channel就会回调相应的操作
    m_channel->setReadCallback(std::bind(&TcpConnection::handleRead, this, std::placeholders::_1));
    m_channel->setWriteCallback(std::bind(&TcpConnection::handleWrite, this));
    m_channel->setCloseCallback(std::bind(&TcpConnection::handleClose, this));
    m_channel->setErrorCallback(std::bind(&TcpConnection::handleError, this));

    LOG_INFO("TcpConnection::construct[%s] at fd = %d \n", m_name.c_str(), sockfd);
    m_socket->setKeepAlive(true);
}

TcpConnection::~TcpConnection()
{
    LOG_INFO("TcpConnection::destruct[%s] at fd = %d  state = %d \n", m_name.c_str(), m_channel->getFd(), (int)m_state);
}

// 发送数据
void TcpConnection::send(const std::string &buf)
{
    if (m_state == kConnected) {
        if (m_loop->isInLoopThread()) {
            sendInLoop(buf.c_str(), buf.size());
        } else {
            m_loop->runInLoop(std::bind(&TcpConnection::sendInLoop, this, buf.c_str(), buf.size()));
        }
    }
}

// 关闭连接
void TcpConnection::shutdown()
{
    if (m_state == kConnected) {
        setState(kDisconnected);
        m_loop->runInLoop(std::bind(&TcpConnection::shutdownInLoop, this));
    }
}

// 建立连接并注册EPOLLIN事件
void TcpConnection::connectEstablished()
{
    setState(kConnected);
    m_channel->tie(shared_from_this());
    m_channel->enableReading(); // 向Poller中注册Channel的EPOLLIN事件

    // 新的连接已建立，执行对应的回调操作
    m_connectionCallback(shared_from_this());
}

// 销毁连接
void TcpConnection::connectDestroyed()
{
    if (m_state == kConnected) {
        setState(kDisconnected);
        m_channel->disableAll(); // 把channel所有感兴趣的事件从Poller中移除
        m_connectionCallback(shared_from_this());
    }

    m_channel->remove(); // 把Channel从Poller中删除
}

void TcpConnection::handleRead(Timestamp receiveTime)
{
    int saveErrno = 0;
    ssize_t n = m_inputBuffer.readFd(m_channel->getFd(), &saveErrno);
    if (n > 0) {
        // 已建立连接的用户，有可读事件发生了，调用用户传入的回调操作onMessage
        m_messageCallback(shared_from_this(), &m_inputBuffer, receiveTime);
    } else if (n == 0) {
        handleClose();
    } else {
        errno = saveErrno;
        LOG_ERROR("TcpConnection::handleRead");
        handleError();
    }
}

void TcpConnection::handleWrite()
{
    if (m_channel->isWriting()) {
        int saveError = 0;
        ssize_t n = m_outputBuffer.writeFd(m_channel->getFd(), &saveError);
        if (n > 0) {
            m_outputBuffer.retrieve(n);
            if (m_outputBuffer.readableBytes() == 0) {
                m_channel->disableWriting();
                if (m_writeCompleteCallback) {
                    // 唤醒m_loop对应的thread线程，执行回调
                    m_loop->queueInLoop(std::bind(m_writeCompleteCallback, shared_from_this()));
                }
                
                if (m_state == kDisconnecting) {
                    shutdownInLoop();
                }
            }
        } else {
            LOG_ERROR("TcpConnection::handlerWrite");
        }
    } else {
        LOG_ERROR("TcpConnection fd = %d is down, no more writing \n", m_channel->getFd());
    }
}

// Poller => channel::closeCallback => TcpConnection::handleClose
void TcpConnection::handleClose()
{
    LOG_INFO("TcpConnection::handleClose fd = %d state = %d \n", m_channel->getFd(), (int)m_state);
    setState(kDisconnected);
    m_channel->disableAll();

    TcpConnectionPtr connPtr(shared_from_this());
    m_connectionCallback(connPtr); // 执行连接关闭的回调
    m_closeCallback(connPtr); // 关闭连接的回调，执行TcpServer::removeConnection回调方法
}

void TcpConnection::handleError()
{
    int optval;
    socklen_t optlen = sizeof(optval);
    int err = 0;
    if (::getsockopt(m_channel->getFd(), SOL_SOCKET, SO_ERROR, &optval, &optlen) < 0) {
        err = errno;
    } else {
        err = optval;
    }
    LOG_ERROR("TcpConnection::handleError name : %s - SO_ERROR : %d \n", m_name.c_str(), err);
}

// 发送数据 应用写的快，而内核发送数据慢，就需要把待发送的数据写入缓冲区，并设置水位回调，防止发送太快
void TcpConnection::sendInLoop(const void *data, size_t len)
{
    ssize_t nwrote = 0;
    size_t remaining = len;
    bool faultError = false;

    // 如果之前调用过改connection的shutdown，则不能再进行数据的发送
    if (m_state == kDisconnected) {
        LOG_ERROR("disconnected, give up writing......");
        return ;
    }

    // m_channel第一次开始写数据，而且缓冲区没有待发送的数据
    if (!m_channel->isWriting() && m_outputBuffer.readableBytes() == 0) {
        nwrote = ::write(m_channel->getFd(), data, len);
        if (nwrote >= 0) {
            remaining = len - nwrote;
            if (remaining == 0 && m_writeCompleteCallback) {
                // 如果已被全部发送完成，就不需要给channel设置epollout事件
                m_loop->queueInLoop(std::bind(m_writeCompleteCallback, shared_from_this()));
            }
        } else {
            // nwrote < 0 出现异常
            nwrote = 0;
            if (errno != EWOULDBLOCK) {
                LOG_ERROR("TcpConnection::sendInLoop");
                // SIGPIPE RESET
                if (errno == EPIPE || errno == ECONNRESET) {
                    faultError = true;
                }
            }
        }
    }

    // 说明当前这一次write并没有把数据全部发送出去，剩余的数据需要保存到缓冲区，然后给channel注册epollout事件
    // poller发现tcp缓冲区有空间就会通知相应的sock -> channel，然后调用m_writeCallback回调方法，也就是调用TcpConnection::handleWrite方法
    // 把发送缓冲区中的数据全部发送完成
    if (!faultError && remaining > 0) {
        // 目前发送缓冲区剩余的待发送数据的长度
        size_t oldLen = m_outputBuffer.readableBytes();
        if (oldLen + remaining >= m_highWaterMark && oldLen < m_highWaterMark && m_highWaterMarkCallback) {
            m_loop->queueInLoop(std::bind(m_highWaterMarkCallback, shared_from_this(), oldLen + remaining));                
        }
        m_outputBuffer.append((char*)data + nwrote, remaining);
        if (!m_channel->isWriting()) {
            m_channel->enableWriting(); // 注册channel的写事件，否则poller不会给channel通知epollout
        }
    }
}


void TcpConnection::shutdownInLoop()
{
    // 说明发送缓冲区的数据已经被发送完毕
    if (!m_channel->isWriting()) {
        m_socket->shutdownWrite(); // 关闭写端
    }
}
